package cn.com.huaxun.base.kafka.config;

import cn.hutool.json.JSONObject;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @descripe：kafka发送配置类
 * @company：南京华讯方舟通讯设备有限公司
 * @author：zhoujinbing
 * @time：2019/1/2 16:45
 * @version：V1.0
 */
@Component
@Slf4j
@AllArgsConstructor
public class KafkaTemplateSender<T> {

    private KafkaTemplate kafkaTemplate;

    /**
     * @description 发送消息
     * @method  send
     * @param  [topic, o]
     * @return void
     * @date: 2019/1/2 16:59
     * @author:zhoujinbing
     */
    public void send(String topic,Object o)
    {
        log.info("---------------kafka 发送消息--------------------");

        ListenableFuture  send = kafkaTemplate.send(topic, new JSONObject(o).toString());
        send.addCallback(new ListenableFutureCallback() {
            @Override
            public void onFailure(Throwable ex) {
                log.info("Produce: The message failed to be sent:" + ex.getMessage());
            }

            @Override
            public void onSuccess(Object result) {
                log.info("Produce: The message sent success:");
            }
        });
    }





}